package com.zhang.sql2;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.util.Collector;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * @title:
 * @author: zhang
 * @date: 2022/3/26 20:49
 */
public class MyTableAggregateFunction1 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //2 创建表 (连接器表)

        String create = "create table clickTable (" +
                " name STRING," +
                " url STRING" +
                " ) WITH ( " +
                " 'connector' = 'filesystem'," +
                " 'path' = 'input/click.txt'," +
                " 'format' = 'csv'" +
                " )";
        tableEnv.executeSql(create);


        //注册自定义函数
        tableEnv.createTemporarySystemFunction("Top2", Top2.class);


        Table aggResult = tableEnv.sqlQuery("" +
                " select name,count(url) cnt " +
                " from clickTable " +
                " group by name");

        Table table = aggResult.groupBy($("name"))
                .flatAggregate(call("Top2", $("cnt")).as("value", "rk"))
                .select( $("value"), $("rk"));

        //转换成流打印输出
        tableEnv.toChangelogStream(table).print();
        env.execute();
    }

    public static class Top2Accum {
        public Long first;
        public Long second;
        public Long oldFirst;
        public Long oldSecond;
    }

    /**
     * The top2 user-defined table aggregate function.
     */
    public static class Top2 extends TableAggregateFunction<Tuple2<Long, Integer>, Top2Accum> {

        @Override
        public Top2Accum createAccumulator() {
            Top2Accum acc = new Top2Accum();
            acc.first = Long.MIN_VALUE;
            acc.second = Long.MIN_VALUE;
            acc.oldFirst = Long.MIN_VALUE;
            acc.oldSecond = Long.MIN_VALUE;
            return acc;
        }

        public void accumulate(Top2Accum acc, Long v) {
            if (v > acc.first) {
                acc.second = acc.first;
                acc.first = v;
            } else if (v > acc.second) {
                acc.second = v;
            }
        }

        public void emitValue(Top2Accum acc, Collector<Tuple2<Long, Integer>> out) {
            // emit the value and rank
            if (acc.first != Long.MIN_VALUE) {
                out.collect(Tuple2.of(acc.first, 1));
            }
            if (acc.second != Long.MIN_VALUE) {
                out.collect(Tuple2.of(acc.second, 2));
            }
        }

        public void emitUpdateWithRetract(Top2Accum acc, RetractableCollector<Tuple2<Long, Integer>> out) {
            if (!acc.first.equals(acc.oldFirst)) {
                // if there is an update, retract old value then emit new value.
                if (acc.oldFirst != Integer.MIN_VALUE) {
                    out.retract(Tuple2.of(acc.oldFirst, 1));
                }
                out.collect(Tuple2.of(acc.first, 1));
                acc.oldFirst = acc.first;
            }

            if (!acc.second.equals(acc.oldSecond)) {
                // if there is an update, retract old value then emit new value.
                if (acc.oldSecond != Integer.MIN_VALUE) {
                    out.retract(Tuple2.of(acc.oldSecond, 2));
                }
                out.collect(Tuple2.of(acc.second, 2));
                acc.oldSecond = acc.second;
            }
        }

        public void retract(Top2Accum acc, Long v){


        }
    }

}
